/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.indexer.hyracks;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.values.IntType;
import edu.ucsd.db.bassarisk.data.values.IntType.IntTools;
import edu.ucsd.db.bassarisk.data.values.IntType.IntegerValueCodec;
import edu.ucsd.db.bassarisk.data.values.IntType.ObjectIntegerReference;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.LongValueCodec;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;

public class MixedTypeSerailizerDeserializer implements
    ISerializerDeserializer<IValueReference<?>> {
  public static final MixedTypeSerailizerDeserializer INSTANCE =
      new MixedTypeSerailizerDeserializer();
  private static final long serialVersionUID = 3256757664385884857L;

  private MixedTypeSerailizerDeserializer() {}

  @Override
  public IValueReference<?> deserialize(final DataInput arg0)
      throws HyracksDataException {
    long typeUid;
    try {
      typeUid = arg0.readLong();
      if (typeUid == 0) {
        return null;
      } else if (typeUid == LongType.INSTANCE.getUid()) {
        return new ObjectLongReference(arg0.readLong());
      } else if (typeUid == IntType.INSTANCE.getUid()) {
        return new ObjectIntegerReference(arg0.readInt());
        // } else if (typeUid == StringType.INSTANCE.getUid()) {
        // return new ObjectStringReference(arg0.readUTF());
      } else {
        String message =
            "Bassarisk type {0} not supported by MixedTypeSerializerDeserializer";
        throw new RuntimeException(String.format(message, typeUid));
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  @SuppressWarnings("unchecked")
  @Override
  public void serialize(final IValueReference<?> arg0, final DataOutput arg1)
      throws HyracksDataException {
    try {
      if (arg0 == null) {
        arg1.writeLong(0);
        return;
      }
      IValueType type = arg0.getType();
      arg1.writeLong(type.getUid());
      if (type == LongType.INSTANCE) {
        IValueReference<LongType> longRef = (IValueReference<LongType>) arg0;
        LongValueCodec longCodec = LongTools.INSTANCE.getGenericCodec();
        arg1.writeLong(longCodec.getLongValue(longRef));
      } else if (type == IntType.INSTANCE) {
        IValueReference<IntType> intRef = (IValueReference<IntType>) arg0;
        IntegerValueCodec intCodec = IntTools.INSTANCE.getGenericCodec();
        arg1.writeInt(intCodec.getIntegerValue(intRef));
        // } else if (type == StringType.INSTANCE) {
        // IValueReference<StringType> stringRef = (IValueReference<StringType>)
        // arg0;
        // IntegerValueCodec stringCodec =
        // StringTools.INSTANCE.getGenericCodec();
        // arg1.writeUTF(stringCodec.getStringValue(stringRef));
      } else {
        String message =
            "Bassarisk type {0} not supported by MixedTypeSerializerDeserializer";
        throw new RuntimeException(String.format(message, type.getClass().getName()));
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

}
